package cn._51doit.flink.day07;

import cn._51doit.flink.day06.KafkaToRedisWordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * 将状态保存到文件系统中（通常使用HFDS、S3）
 *
 * 该方式已经过时
 *
 * 1.首先下载flink和hadoop3整合的依赖
 *   https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop-3-uber
 * 2.需要将jar包下载，然后放到flink的的按照目录的lib下
 * 3.如果是standalone模式，需要重启集群，如果是on yarn不需要
 *
 */
public class FsStateBackendDemo {

    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //开启checkpoint
        env.enableCheckpointing(30000);
        //设置statebackend
        env.setStateBackend(new FsStateBackend("hdfs://node-1.51doit.cn:9000/myck"));

        Properties properties = new Properties();

        properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092");
        properties.setProperty("group.id", "test888");
        properties.setProperty("auto.offset.reset", "earliest"); //如果没有记录历史偏移量就从头读

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>(
                "words",
                new SimpleStringSchema(),
                properties
        );

        DataStreamSource<String> lines = env.addSource(kafkaConsumer);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void flatMap(String in, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = in.split(" ");
                for (String word : words) {
                    collector.collect(Tuple2.of(word, 1));
                }
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndOne.keyBy(t -> t.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> res = keyed.sum(1);

        //redis的配置
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost("node-3.51doit.cn")
                .setDatabase(3)
                .setPassword("123456")
                .build();
        //传入redis sink和配置mapper
        res.addSink(new RedisSink<Tuple2<String, Integer>>(conf, new KafkaToRedisWordCount.RedisWordCountMapper()));

        //5.执行并挂起
        env.execute();

    }


    /**
     * Redis的映射类，决定以何种方式写入到Redis中，那个字段作为key，哪个字段作为value
     * 泛型为输入到Sink中的数据类型
     */
    private static class RedisWordCountMapper implements RedisMapper<Tuple2<String, Integer>> {

        /**
         * 决定以何种方式写入到redis中
         * @return
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            //大key -> (小key， 小value)
            return new RedisCommandDescription(RedisCommand.HSET, "wc");
        }

        /**
         * 设置小key的内容
         * @param data
         * @return
         */
        @Override
        public String getKeyFromData(Tuple2<String, Integer> data) {
            return data.f0;
        }

        /**
         * 设置小value的内容
         * @param data
         * @return
         */
        @Override
        public String getValueFromData(Tuple2<String, Integer> data) {
            return data.f1.toString();
        }
    }

}
